package mapreduceGroup;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;

/**
 * 分组操作，当第一列不同时，升序，当第一列相同时，第二列升序
 * 
 * @author user
 *
 */
public class MapreduceGroup {
	static final String INPUT_PATH = "hdfs://192.168.8.70:8020/root/kxdinput";
	static final String OUT_PUT = "hdfs://192.168.8.70:8020/root/kxdoutput";

	public static void main(String[] args) throws IllegalArgumentException, IOException, URISyntaxException,
			ClassNotFoundException, InterruptedException {
		final Configuration configuration = new Configuration();
		final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), configuration);
		if (fileSystem.exists(new Path(OUT_PUT))) {
			fileSystem.delete(new Path(OUT_PUT), true);
		}
		final Job job = new Job(configuration, MapreduceGroup.class.getSimpleName());

		job.setInputFormatClass(TextInputFormat.class);
		job.setMapperClass(MyMapper.class);
		job.setMapOutputKeyClass(NewK2.class);
		job.setMapOutputValueClass(LongWritable.class);
		job.setPartitionerClass(HashPartitioner.class);
		job.setNumReduceTasks(1);
		FileInputFormat.addInputPath(job, new Path(INPUT_PATH));
		job.setGroupingComparatorClass(MyGroupingComparator.class);

		job.setReducerClass(MyReducer.class);
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(LongWritable.class);

		FileOutputFormat.setOutputPath(job, new Path(OUT_PUT));

		job.setOutputFormatClass(TextOutputFormat.class);

		job.waitForCompletion(true);

	}

	static class MyMapper extends Mapper<LongWritable, Text, NewK2, LongWritable> {
		protected void map(LongWritable key, Text value,
				org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, NewK2, LongWritable>.Context context)
				throws IOException, InterruptedException {
			final String[] splitted = value.toString().split("\t");
			final NewK2 k2 = new NewK2(Long.parseLong(splitted[0]), Long.parseLong(splitted[1]));
			final LongWritable v2 = new LongWritable(Long.parseLong(splitted[1]));
			context.write(k2, v2);

		};
	}

	static class MyReducer extends Reducer<NewK2, LongWritable, LongWritable, LongWritable> {
		@Override
		protected void reduce(NewK2 k2, Iterable<LongWritable> v2s,
				Reducer<NewK2, LongWritable, LongWritable, LongWritable>.Context context)
				throws IOException, InterruptedException {
			long min = Long.MAX_VALUE;
			for (LongWritable v2 : v2s) {
				if (v2.get() < min) {
					min = v2.get();
				}
			}
			context.write(new LongWritable(k2.first), new LongWritable(min));
		}
	}

	static class NewK2 implements WritableComparable<NewK2> {
		Long first;
		Long second;

		public NewK2() {

		}

		public NewK2(long first, long second) {
			this.first = first;
			this.second = second;
		}

		@Override
		public void write(DataOutput out) throws IOException {
			out.writeLong(first);
			out.writeLong(second);

		}

		@Override
		public void readFields(DataInput in) throws IOException {

			this.first = in.readLong();
			this.second = in.readLong();

		}

		@Override
		public int compareTo(NewK2 o) {
			final long minus = this.first - o.first;
			if (minus != 0) {
				return (int) minus;
			}
			return (int) (this.second - o.second);
		}

		@Override
		public int hashCode() {
			return this.first.hashCode() + this.second.hashCode();
		}

		@Override
		public boolean equals(Object obj) {
			if (!(obj instanceof NewK2)) {
				return false;
			}
			NewK2 ok2 = (NewK2) obj;
			return (this.first == ok2.first) && (this.second == ok2.second);
		}
	}

	/**
	 * 自定义分组比较器
	 */
	static class MyGroupingComparator implements RawComparator<NewK2> {

		@Override
		public int compare(NewK2 o1, NewK2 o2) {
			return (int) (o1.first - o2.first);
		}

		/**
		 * @param arg0
		 *            表示第一个参与比较的字节数组
		 * @param arg1
		 *            表示第一个参与比较的字节数组的起始位置
		 * @param arg2
		 *            表示第一个参与比较的字节数组的偏移量
		 * 
		 * @param arg3
		 *            表示第二个参与比较的字节数组
		 * @param arg4
		 *            表示第二个参与比较的字节数组的起始位置
		 * @param arg5
		 *            表示第二个参与比较的字节数组的偏移量
		 */
		@Override
		public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
			return WritableComparator.compareBytes(b1, s1, l1, b2, s2, l2);
		}

	}
}
